package cn.spark.study.sql;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;

/**
 * 每日UV数据
 *
 * @author jun.zhang6
 * @date 2020/11/15
 */
public class DailyUV {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setMaster("local").setMaster("DailyUV");

        JavaSparkContext sc = new JavaSparkContext(conf);

        SQLContext sqlContext = new SQLContext(sc);

        List<String> userAccessLog = new LinkedList<String>();
        userAccessLog.add("2015-10-01,1122");
        userAccessLog.add("2015-10-01,1122");
        userAccessLog.add("2015-10-01,1123");
        userAccessLog.add("2015-10-01,1124");
        userAccessLog.add("2015-10-02,1122");
        userAccessLog.add("2015-10-02,1121");
        userAccessLog.add("2015-10-02,1123");
        userAccessLog.add("2015-10-02,1123");

        JavaRDD<String> userAccessLogRDD = sc.parallelize(userAccessLog);

        JavaRDD<Row> userAccessLogRowRDD = userAccessLogRDD.map(new Function<String, Row>() {
            @Override
            public Row call(String log) throws Exception {
                return RowFactory.create(log.split(",")[0], Integer.parseInt(log.split(",")[1]));
            }
        });

        //构造DataStruct
        List<StructField> structFields = new ArrayList<StructField>();
        structFields.add(DataTypes.createStructField("date", DataTypes.StringType, true));
        structFields.add(DataTypes.createStructField("userId", DataTypes.IntegerType, true));
        StructType structType = DataTypes.createStructType(structFields);
        DataFrame userAccessLogRowDF = sqlContext.createDataFrame(userAccessLogRowRDD, structType);

        userAccessLogRowDF.groupBy("data");

    }
}
